﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using XcoAppSpaces.Core.Ports.Remoting;
using XcoAppSpaces.Core.Ports.Utils;

namespace XcoAppSpaces.Core.Ports
{
	/// <summary>
	/// Abstract port base class, defines default PortReceiverOptions, some static methods for creating ports.
	/// </summary>
	public abstract class Port : IRemotablePort
	{
		/// <summary>
		/// Default PortReceiverOptions, which are used whenever a receiver is registered and no options are directly specified.
		/// By default, the MaxDegreeOfParallelism equal the number of cores, or at least 2. There is no specific task scheduler
		/// set by default.
		/// If you want to have other default settings, you can change these settings before instantiating your ports.
		/// </summary>
		public static readonly PortReceiverOptions DefaultOptions = new PortReceiverOptions
		{
			MaxDegreeOfParallelism = Math.Max(2, Environment.ProcessorCount),
			TaskScheduler =  null //uses the default scheduler
		};

		/// <summary>
		/// Creates a port using the given receiver.
		/// </summary>
		/// <typeparam name="T">The port's message type.</typeparam>
		/// <param name="receiver">The receiver that should process messages from the port.</param>
		/// <returns>The created port instance.</returns>
		public static Port<T> WithReceiver<T>(Action<T> receiver)
		{
			return new Port<T>(receiver);
		}

		/// <summary>
		/// Creates a port using the given receiver.
		/// </summary>
		/// <typeparam name="T">The port's message type.</typeparam>
		/// <param name="receiver">The receiver that should process messages from the port.</param>
		/// <param name="options">Processing options, like the MaxDegreeOfParallelism.</param>
		/// <returns>The created port instance.</returns>
		public static Port<T> WithReceiver<T>(Action<T> receiver, PortReceiverOptions options)
		{
			return new Port<T>(receiver, options);
		}

		/// <summary>
		/// Creates a port using the given receiver.
		/// </summary>
		/// <typeparam name="T">The port's message type.</typeparam>
		/// <param name="receiver">The receiver that should process messages from the port.</param>
		/// <returns>The created port instance.</returns>
		public static Port<T> WithReceiver<T>(Func<T, Task> receiver)
		{
			return new Port<T>(receiver);
		}

		/// <summary>
		/// Creates a port using the given receiver.
		/// </summary>
		/// <typeparam name="T">The port's message type.</typeparam>
		/// <param name="receiver">The receiver that should process messages from the port.</param>
		/// <param name="options">Processing options, like the MaxDegreeOfParallelism.</param>
		/// <returns>The created port instance.</returns>
		public static Port<T> WithReceiver<T>(Func<T, Task> receiver, PortReceiverOptions options)
		{
			return new Port<T>(receiver, options);
		}

		/// <summary>
		/// Creates a port using the given receiver. The items will be processed sequentially.
		/// </summary>
		/// <typeparam name="T">The port's message type.</typeparam>
		/// <param name="receiver">The receiver that should process messages from the port.</param>
		/// <returns>The created port instance.</returns>
		public static Port<T> WithSeqentialReceiver<T>(Action<T> receiver)
		{
			return new Port<T>(receiver, new PortReceiverOptions { MaxDegreeOfParallelism = 1, TaskScheduler = DefaultOptions.TaskScheduler });
		}

		/// <summary>
		/// Creates a port using the given receiver. The items will be processed sequentially.
		/// </summary>
		/// <typeparam name="T">The port's message type.</typeparam>
		/// <param name="receiver">The receiver that should process messages from the port.</param>
		/// <returns>The created port instance.</returns>
		public static Port<T> WithSeqentialReceiver<T>(Func<T, Task> receiver)
		{
			return new Port<T>(receiver, new PortReceiverOptions { MaxDegreeOfParallelism = 1, TaskScheduler = DefaultOptions.TaskScheduler });
		}

		/// <summary>
		/// Returns a task that completes as soon as an item from both ports was received. The result is given as a tuple containing both items.
		/// </summary>
		/// <typeparam name="T1">The first port's message type.</typeparam>
		/// <typeparam name="T2">The second port's message type.</typeparam>
		/// <param name="port1">The first port.</param>
		/// <param name="port2">The second port.</param>
		/// <returns>The task to wait for the result.</returns>
		public static async Task<Tuple<T1, T2>> ReceiveFromAll<T1, T2>(IPort<T1> port1, IPort<T2> port2)
		{
			var t1 = port1.Receive();
			var t2 = port2.Receive();
			await Task.WhenAll(t1, t2).ConfigureAwait(false);
			return Tuple.Create(t1.Result, t2.Result);
		}
		/// <summary>
		/// Returns a task that returns as completes as an item from one of the two ports is received. The result is given as a tuple
		/// containing the received item for one port, and the default value (null) for the other port.
		/// </summary>
		/// <typeparam name="T1">The first port's message type.</typeparam>
		/// <typeparam name="T2">The second port's message type.</typeparam>
		/// <param name="port1">The first port.</param>
		/// <param name="port2">The second port.</param>
		/// <returns>The task to wait for the result.</returns>
		public static async Task<Tuple<T1, T2>> ReceiveFromAny<T1, T2>(IPort<T1> port1, IPort<T2> port2)
		{
			var t1 = port1.Receive();
			var t2 = port2.Receive();
			await Task.WhenAny(t1, t2).ConfigureAwait(false);
			if (t1.IsCompleted)
				return Tuple.Create(t1.Result, default(T2));
			return Tuple.Create(default(T1), t2.Result);
		}

		/// <summary>
		/// This field can be set to mark the port as a remote port. It gives access to remote port related infos
		/// like the remote address.
		/// </summary>
		public XcoPortInfo RemotePortInfo { get; set; }

		internal abstract void MapToConcurrentExclusiveGroup(AsyncReaderWriterLock asyncLock, bool writer);
	}


	/// <summary>
	/// The port class  is the main element of parallelization and asynchronous communication in the XcoAppSpaces.
	/// Messages can be posted to the port, and either received or processed ansychronously.
	/// </summary>
	/// <typeparam name="T">The port's message type.</typeparam>
	public class Port<T> : Port, IPort<T>, IPortBypass, IRemotedPost
	{
		//the port buffer does the actual scheduling and item storing work
		private readonly PortBufferCQ<PortElement<T>> _buffer = new PortBufferCQ<PortElement<T>>();
		private Action<T> _receiverAction;
		private Func<T, Task> _receiverFunc;
		private AsyncReaderWriterLock _asyncLock;
		private bool _writer;

		private static readonly Task completedTask = Task.FromResult<object>(null);

		/// <summary>
		/// Creates a new port without a receiver (will store posted items, until their are either received
		/// using the Receive() method, or a receiver is installed.
		/// </summary>
		public Port()
		{
		}

		/// <summary>
		/// Creates a new port, using the given receiver for processing posted items.
		/// </summary>
		/// <param name="receiver">The receiver that should be used for processing items.</param>
		public Port(Action<T> receiver)
		{
			RegisterReceiver(receiver);
		}

		/// <summary>
		/// Creates a new port, using the given receiver for processing posted items.
		/// </summary>
		/// <param name="receiver">The receiver that should be used for processing items.</param>
		/// <param name="options">Processing options, like the MaxDegreeOfParallelism.</param>
		public Port(Action<T> receiver, PortReceiverOptions options)
		{
			RegisterReceiver(receiver, options);
		}

		/// <summary>
		/// Creates a new port, using the given receiver for processing posted items.
		/// </summary>
		/// <param name="receiver">The receiver that should be used for processing items.</param>
		public Port(Func<T, Task> receiver)
		{
			RegisterReceiver(receiver);
		}

		/// <summary>
		/// Creates a new port, using the given receiver for processing posted items.
		/// </summary>
		/// <param name="receiver">The receiver that should be used for processing items.</param>
		/// <param name="options">Processing options, like the MaxDegreeOfParallelism.</param>
		public Port(Func<T, Task> receiver, PortReceiverOptions options)
		{
			RegisterReceiver(receiver, options);
		}

		/// <summary>
		/// The number of items that are stored in the port.
		/// </summary>
		public int ItemCount
	    {
			get { return _buffer.Count; }
	    }

		/// <summary>
		/// The type of messages that can be posted to the port.
		/// </summary>
		public Type MessageType { get { return typeof (T); } }


		/// <summary>
		/// Posts an item to the port. If the message doesn't correspond to the port's type, an exception is thrown.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		public void PostUnknownType(object item)
		{
			PostUnknownType(item, null, null);
		}

		/// <summary>
		/// Posts an item to the port. If the message doesn't correspond to the port's type, an exception is thrown.
		/// The given error port is used as causality, when exceptions are thrown.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <param name="errorPort">The error port to be used when exceptions are thrown.</param>
		public void PostUnknownType(object item, IPort<Exception> errorPort)
		{
			PostUnknownType(item, errorPort, null);
		}

		/// <summary>
		/// Posts an item to the port. If the message doesn't correspond to the port's type, an exception is thrown.
		/// The given error and ack ports are used as causality.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <param name="errorPort">The error port to be used when exceptions are thrown.</param>
		/// <param name="ackPort">The port where an ack message should be posted.</param>
		public void PostUnknownType(object item, IPort<Exception> errorPort, IPort<Ack> ackPort)
		{
			if (!(item is T))
				throw new ArgumentException("This port can only take messages of type " + typeof(T));
			Post((T)item, errorPort, ackPort);
		}

		/// <summary>
		/// Posts an item to the port and returns a task that completes as soon as the item was sent (or goes
		/// into faulted state when sending fails). If the port is not remote, then a task will be returned
		/// that is already completed. If the message doesn't correspond to the port's type, an exception is thrown.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <returns>A task to wait for the item to be sent.</returns>
		public Task PostUnknownTypeAndAwaitSent(object item)
		{
			return PostUnknownTypeAndAwaitSent(item, null);
		}

		/// <summary>
		/// Posts an item to the port and returns a task that completes as soon as the item was sent (or goes
		/// into faulted state when sending fails). If the port is not remote, then a task will be returned
		/// that is already completed. If the message doesn't correspond to the port's type, an exception is thrown.
		/// Since communication errors are returned to the task, the given error port does not react to
		/// communication errors, but only to errors happening later when the message is being processed.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <param name="errorPort">Error port for receiving exceptions that happen during message processing.</param>
		/// <returns>A task to wait for the item to be sent.</returns>
		public Task PostUnknownTypeAndAwaitSent(object item, IPort<Exception> errorPort)
		{
			if (!(item is T))
				throw new ArgumentException("This port can only take messages of type " + typeof(T));
			return PostAndAwaitSent((T)item, errorPort);
		}

		/// <summary>
		/// Posts an item to the port.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		public void Post(T item)
		{
			Post(item, null, null);
		}

		/// <summary>
		/// Posts an item to the port. The given error port is used as causality, when exceptions are thrown.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <param name="errorPort">The error port to be used when exceptions are thrown.</param>
		public void Post(T item, IPort<Exception> errorPort)
		{
			Post(item, errorPort, null);
		}

		/// <summary>
		/// Posts an item to the port. The given error and ack ports are used as causality.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <param name="errorPort">The error port to be used when exceptions are thrown.</param>
		/// <param name="ackPort">The port where an ack message should be posted.</param>
		public void Post(T item, IPort<Exception> errorPort, IPort<Ack> ackPort)
		{
			var element = new PortElement<T> { Item = item, Causalities = CollectCausalities(errorPort, ackPort)};
			_buffer.Post(element);
		}

		/// <summary>
		/// Posts an item to the port and returns a task that completes as soon as the item was sent (or goes
		/// into faulted state when sending fails). If the port is not remote, then a task will be returned
		/// that is already completed.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <returns>A task to wait for the item to be sent.</returns>
		public Task PostAndAwaitSent(T item)
		{
			return PostAndAwaitSent(item, null);
		}

		/// <summary>
		/// Posts an item to the port and returns a task that completes as soon as the item was sent (or goes
		/// into faulted state when sending fails). If the port is not remote, then a task will be returned
		/// that is already completed. If the message doesn't correspond to the port's type, an exception is thrown.
		/// Since communication errors are returned to the task, the given error port does not react to
		/// communication errors, but only to errors happening later when the message is being processed.
		/// </summary>
		/// <param name="item">The item to be posted.</param>
		/// <param name="errorPort">Error port for receiving exceptions that happen during message processing.</param>
		/// <returns>A task to wait for the item to be sent.</returns>
		public Task PostAndAwaitSent(T item, IPort<Exception> errorPort)
		{
			bool isRemotePort = this.IsRemotePort();
			var element = new PortElement<T> { Item = item, Causalities = CollectCausalities(errorPort, null), 
				SentCompletionSource = (isRemotePort ? new TaskCompletionSource<object>() : null)};
			_buffer.Post(element);
			return (isRemotePort ? element.SentCompletionSource.Task : completedTask); //only items from a remote port are sent - for all others return an already completed task
		}

		private List<ICausality> CollectCausalities(IPort<Exception> errorPort, IPort<Ack> ackPort)
		{
			//collects the causalities that are currently active
			var activeCausalities = CausalityContext.ActiveCausalities;
			if (activeCausalities == null && errorPort == null && ackPort == null)
				return null;
			var causalities = new List<ICausality>();
			if (activeCausalities != null)
				causalities.AddRange(activeCausalities);
			if (errorPort != null || ackPort != null)
				causalities.Add(new Causality(errorPort, ackPort));
			return causalities;
		}

		/// <summary>
		/// Asynchronously receives an item from the port, with infinite timeout.
		/// </summary>
		/// <returns>A task for asynchronously receiving an item from the port.</returns>
		public Task<T> Receive()
		{
			return Receive(0);
		}

		/// <summary>
		/// Asynchronously receives an item from the port, with the specified timeout.
		/// </summary>
		/// <param name="timeout">The maximum time in milliseconds to wait for an item.</param>
		/// <returns>A task for asynchronously receiving an item from the port.</returns>
		public Task<T> Receive(int timeout)
		{
			//strangely, the variant using await doesn't seem to work on Android, when the operation is running into a timeout
			return _buffer.ReceiveAsync(timeout).ContinueWith(t =>
				{
					if (t.IsFaulted)
					{
						throw t.Exception.InnerExceptions.First();
					}
					return t.Result.Item;
				}, TaskContinuationOptions.ExecuteSynchronously);

			//var value = await _buffer.ReceiveAsync(timeout);
			//value.SignalCompletion();
			//return value.Item;
		}

		/// <summary>
		/// Removes all items from the port.
		/// </summary>
		public void Clear()
		{
			_buffer.Clear();
		}

		/// <summary>
		/// Registers a receiver which should be used for processing items that are posted to the port.
		/// </summary>
		/// <param name="receiver">The receiver action.</param>
		public void RegisterReceiver(Action<T> receiver)
		{
			RegisterReceiver(receiver, null);
		}

		/// <summary>
		/// Registers a receiver which should be used for processing items that are posted to the port.
		/// </summary>
		/// <param name="receiver">The receiver action.</param>
		/// <param name="options">Processing options, like MaxDegreeOfParallelism.</param>
		public void RegisterReceiver(Action<T> receiver, PortReceiverOptions options)
		{
			_receiverAction = receiver;
			_buffer.RegisterReceiver((Action<PortElement<T>>)ProcessItemAction, options ?? DefaultOptions);
		}

		/// <summary>
		/// Registers a receiver which should be used for processing items that are posted to the port.
		/// This kind of receiver allows the use of async/await.
		/// </summary>
		/// <param name="receiver">The receiver function.</param>
		public void RegisterReceiver(Func<T, Task> receiver)
		{
			RegisterReceiver(receiver, null);
		}

		/// <summary>
		/// Registers a receiver which should be used for processing items that are posted to the port.
		/// This kind of receiver allows the use of async/await.
		/// </summary>
		/// <param name="receiver">The receiver function.</param>
		/// <param name="options">Processing options, like MaxDegreeOfParallelism.</param>
		public void RegisterReceiver(Func<T, Task> receiver, PortReceiverOptions options)
		{
			_receiverFunc = receiver;
			_buffer.RegisterReceiver((Func<PortElement<T>, Task>)ProcessItemFunc, options ?? DefaultOptions);
		}

		/// <summary>
		/// Registers a receiver which should be used for processing items that are posted to the port.
		/// An exception is thrown if the delegate is not usable for the port's message type.
		/// </summary>
		/// <param name="method">The receiver delegate.</param>
		public void RegisterUnknownTypeReceiver(Delegate method)
		{
			RegisterUnknownTypeReceiver(method, null);
		}

		/// <summary>
		/// Registers a receiver which should be used for processing items that are posted to the port.
		/// An exception is thrown if the delegate is not usable for the port's message type.
		/// </summary>
		/// <param name="method">The receiver delegate.</param>
		/// <param name="options">Processing options, like MaxDegreeOfParallelism.</param>
		public void RegisterUnknownTypeReceiver(Delegate method, PortReceiverOptions options)
		{
			if (method is Action<T>)
				RegisterReceiver((Action<T>)method, options);
			else if (method is Func<T, Task>)
				RegisterReceiver((Func<T, Task>)method, options);
			else if (method is Action<object>)
				RegisterReceiver(_ =>  ((Action<object>)method)(_), options);
			else if (method is Func<object, Task>)
				RegisterReceiver(_ => ((Func<object, Task>)method)(_), options);
			else
				throw new ArgumentException("Cannot register delegate " + method + " at a port with message type " + MessageType);
		}

		/// <summary>
		/// Unregisters the currently installed receiver.
		/// </summary>
		public void UnregisterReceiver()
		{
			_buffer.UnregisterReceiver();
		}

		/// <summary>
		/// Defines that the port is part of a ConcurrentExclusiveGroup, and therefore needs to acquire a lock
		/// before processing an item. Only a port with a registered receiver can be added to a ConcurrentExclusiveGroup.
		/// </summary>
		/// <param name="asyncLock">The AsyncReaderWriterLock object from which a lock must be acquired when an item is processed.</param>
		/// <param name="writer">If true the port is a writer (exclusive) port. If false, it is a reader (concurrent) port.</param>
		internal override void MapToConcurrentExclusiveGroup(AsyncReaderWriterLock asyncLock, bool writer)
		{
			if (_receiverAction == null && _receiverFunc == null)
				throw new Exception("cannot use a port in a concurrent/exclusive group that has no receiver");
			_asyncLock = asyncLock;
			_writer = writer;
		}

		/// <summary>
		/// Processing method for an action receiver. Acquires a lock if part of a ConcurrentExclusiveGroup, and signals error or completion.
		/// </summary>
		/// <param name="item">The item to be processed.</param>
		private void ProcessItemAction(PortElement<T> item)
		{
			try
			{
				if (_asyncLock != null)
					ProcessItemActionInGroup(item);
					//this is in a separate method because it is the only part of the method that requires async/await
				else
				{
					InstallContext(item);
					_receiverAction(item.Item);
				}
			}
			catch (Exception ex)
			{
				item.SignalError(ex);
			}
			finally
			{
				UninstallContext(item);
			}
		}

		private static void UninstallContext(IPortElement item)
		{
			CausalityContext.Clear();
			if (item.MessageContext != null)
				MessageContext.Current = null;
		}

		private static void InstallContext(IPortElement item)
		{
			if (item.Causalities != null)
				CausalityContext.AddAll(item.Causalities);
			MessageContext.Current = item.MessageContext ?? new MessageContext{IsRemoteMessage = false};
		}

		private async void ProcessItemActionInGroup(PortElement<T> item)
		{
			using (await (_writer ? _asyncLock.WriterLockAsync() : _asyncLock.ReaderLockAsync()).ConfigureAwait(false))
			{
				InstallContext(item);
				_receiverAction(item.Item);
			}
		}

		/// <summary>
		/// Processing method for an async function receiver (is separated from the action processing method, because it
		/// is more complicated and requires to await the processing method. 
		/// Acquires a lock if part of a ConcurrentExclusiveGroup, and signals error or completion.
		/// </summary>
		/// <param name="item">The item to be processed.</param>
		private async Task ProcessItemFunc(PortElement<T> item)
		{
			try
			{
				//PROBLEM: it cannot be guaranteed that the causalities are sustained if await is used!!
				//(another task could use the thread and clear the causalities...)
				//maybe use CallContext, but this doesn't seem to be available in a PCL
				// -> http://blog.stephencleary.com/2013/04/implicit-async-context-asynclocal.html
				
				if (_asyncLock != null)
				{
					using (await (_writer ? _asyncLock.WriterLockAsync() : _asyncLock.ReaderLockAsync()))
					{
						InstallContext(item);
						await _receiverFunc(item.Item);
					}
				}
				else
				{
					InstallContext(item);
					await _receiverFunc(item.Item);
				}
			}
			catch (Exception ex)
			{
				item.SignalError(ex);
			}
			finally
			{
				UninstallContext(item);
			}
		}

		/// <summary>
		/// Registers a receiver that bypasses most of the normal port logic (like exception handling), and allows direct access to the PortElement instance.
		/// </summary>
		/// <param name="bypassAction">The action to be called.</param>
		/// <param name="options">Receiver options.</param>
		void IPortBypass.RegisterBypassReceiver(Action<IPort, IPortElement> bypassAction, PortReceiverOptions options)
		{
			_buffer.RegisterReceiver(pe => bypassAction(this,pe), options);
		}

		void IRemotedPost.PostRemoted(object item, IEnumerable<ICausality> causalities, MessageContext msgContext)
		{
			if (!(item is T))
				throw new ArgumentException("This port can only take messages of type " + typeof(T));
			var element = new PortElement<T> { Item = (T)item, Causalities = (causalities != null ? causalities.ToList() : null), MessageContext = msgContext };
			_buffer.Post(element);
		}
	}
}
